package com.supertool.tong.merger.util;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.lang.System;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.util.ReflectionUtils;


import com.supertool.tong.merger.util.MzSequenceFile;


public class MergeRecordReader extends RecordReader<Comparable,Text>{

    private CombineFileSplit split;
    private long totalOffset = 0;
    private long totalLength = 0;
    private List<ReadStatus> statusList = new LinkedList<ReadStatus>();
    private Comparable key = null;
    private Text value = null;
    private LogParser logParser=null;

    MergeRecordReader(Configuration conf) {
        try {
            System.out.println("==================================");
            System.out.println(conf.get("tong.parser.class"));
            System.out.println("==================================");
            logParser=(LogParser) ReflectionUtils.newInstance(
                    conf.getClassByName(conf.get("tong.parser.class")), conf);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            logParser=new LogParser();
        }catch (NullPointerException e){
            e.printStackTrace();
            logParser=new LogParser();
        }
    }


    private static class ReadStatus {
        //private FSDataInputStream stream;
        //private BufferedReader reader;
        MzSequenceFile.Reader seqReader;
        private Path path;
        private Comparable key;
        private String value;
        private long length;
    }

    private boolean streamNext(ReadStatus one) throws IOException{
        for (;;) {
            //long op = one.stream.getPos();
            long op = one.seqReader.getPos();
			/*String line = one.reader.readLine();
			if (line == null) {
				one.reader.close();
				return false;

			}*/
            Text line = new Text();
            if(!one.seqReader.next(line)){
                one.seqReader.close();
                return false;
            }

            //long np = one.stream.getPos();
            long np = one.seqReader.getPos();
            one.key=logParser.parse(line.toString());
            if (one.key!=null) {
                one.value = line.toString();
                one.length = np - op + 1;
                return true;
            }else{
                //TODO throw exception
                // System.out.println("")
            }
        }
    }

    @Override
    public void close() throws IOException {
        for (ReadStatus status : statusList) {
            //status.stream.close();
            status.seqReader.close();
        }
    }

    @Override
    public float getProgress() throws IOException {
        return ((float) this.totalOffset) / totalLength;
    }

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        Configuration job = context.getConfiguration();
        this.split = (CombineFileSplit) split;
        int i = 0;
        int buffersize = 4 * 1024*1024;
        for (Path path : this.split.getPaths()) {
            ReadStatus one = new ReadStatus();
            FileSystem fs = path.getFileSystem(job);
			/*one.stream = fs.open(path);
			one.reader = new BufferedReader(new InputStreamReader(
					one.stream), buffersize);*/
            one.seqReader = new MzSequenceFile.Reader(fs,path,fs.getFileStatus(path).getLen(),job,null);
            one.path = path;
            totalLength += this.split.getLength(i);
            if (streamNext(one)) {
                statusList.add(one);
            }
            i++;
        }
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        Comparable minKey = null;
        ReadStatus minStatus = null;
        for (ReadStatus status : statusList) {
            if (minKey == null) {
                minKey = status.key;
                minStatus = status;
                continue;
            }
            if (minKey.compareTo(status.key)>0) {
                minKey = status.key;
                minStatus = status;
            }
        }
        if (minKey == null) {
            return false;
        }
        totalOffset += minStatus.length;
        key=minKey;
        if(value==null)
            value = new Text();
        value.set(minStatus.value);
        if (!streamNext(minStatus)) {
            statusList.remove(minStatus);
        }
        return true;
    }

    @Override
    public Comparable getCurrentKey() throws IOException, InterruptedException {
        return this.key;
    }

    @Override
    public Text getCurrentValue() throws IOException, InterruptedException {
        return this.value;
    }
}
